package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;

/**
 * 封装维度异步查询的函数类DimAsyncFunction
 * 该类继承异步方法类RichAsyncFunction，实现自定义维度查询接口
 * 其中RichAsyncFunction<IN,OUT>是Flink提供的异步方法类，此处因为是查询操作输入类和返回类一致，所以是<T,T>。
 *    RichAsyncFunction这个类要实现两个方法:
 *    open用于初始化异步连接池。
 *    asyncInvoke方法是核心方法，里面的操作必须是异步的，如果你查询的数据库有异步api也可以用线程的异步方法，如果没有异步方法，就要自己利用线程池等方式实现异步查询。
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements  DimJoinFunction<T>{
    private ExecutorService executorService;
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        executorService = ThreadPoolUtil.getInstance();
    }

    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        //开启多个线程
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                //发送异步请求，完成维度关联
                try {
                    long start = System.currentTimeMillis();
                    //1.根据流中处理的元素obj，获取要关联的维度的主键
                    String key = getKey(obj);
                    //2.根据维度主键获取维度对象
                    JSONObject dimInfoJsonObj = DimUtil.getDimInfo(tableName, key);
                    //3.根据维度获取相关的维度属性  补充到流中处理的元素的obj上
                    if (dimInfoJsonObj != null && dimInfoJsonObj.size()>0){
                        join(dimInfoJsonObj,obj);
                    }
                    Long end = System.currentTimeMillis();
                    //获取数据库交互的结果并发送给 ResultFuture的回调函数
                    resultFuture.complete(Collections.singleton(obj));
                    System.out.println("异步维度关联耗时:" +(end - start) + "毫秒");
                } catch (Exception e) {
                    e.printStackTrace();
                    throw  new RuntimeException("~~~异步关联维度数据失败~~~");
                }
            }
        });
    }
}
